package com.codeduck.kafka;

import com.codeduck.kafka.consumer.IMessageConsumer;
import com.codeduck.kafka.producer.KafkaProducerFactory;
import com.codeduck.kafka.producer.ProducerProperties;
import com.codeduck.pojo.TopicHandler;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author: <a href="mailto:codeduck@163.com">Codeduck</a>
 * @date: 2023/6/23
 */
@Component
public class KafkaInvoker<K,V> extends KafkaOperations<K, V> {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaInvoker.class);

    private KafkaProducerFactory<K,V> producerFactory;

    @Resource
    private IMessageConsumer messageConsumer;

    public KafkaInvoker() {
        this.producerFactory = getProducerFactory();
    }

    public void send(String topic, @Nullable V data) {
        ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
        doSend(producerRecord);
    }

    public void send(String topic, K key, @Nullable V data) {
        ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, data);
        doSend(producerRecord);
    }

    public void send(String topic, Integer partition, K key, @Nullable V data) {
        ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
        doSend(producerRecord);
    }

    public void send(ProducerRecord<K, V> record) {
        doSend(record);
    }

    private void doSend(ProducerRecord<K, V> record) {
        final Producer<K, V> producer = this.producerFactory.createProducer();
        producer.send(record);
        // 是否手动刷新
        this.LOGGER.info("Sent: " + record);
    }

    public KafkaProducerFactory getProducerFactory() {
        Map<String, Object> configMap = new HashMap<>();
        ProducerProperties.getProperties().forEach((k, v) -> {
            configMap.put((String) k,v);
        });
        return new KafkaProducerFactory(configMap);
    }

    public void subscribe(Properties properties, TopicHandler topicHandler) {
        messageConsumer.subscribe(properties, topicHandler);
    }
}
